package com.lizhiyu.flink.demo4_transformation;

import com.lizhiyu.flink.model.VideoOrder;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.Date;
import java.util.Objects;


/***
 *  对信息进行分组，然后求和
 *  场景:每种商品的总销售金额的实时计算
 *  1、对订单信息，先对订单信息根据标题分组，然后根据每个组中的信息求和
 *
 *  keyBy后使用 sum 进行求和,sum 仅仅能对一些pojo的固定字段，或者Tuple中指定位置的字段进行求和，不能进行一些自定义的操作，reduce则能
 */
public class KeyByTransformMation {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<VideoOrder> ds = env.fromCollection(
                Arrays.asList(
                        new VideoOrder("1", "java ", 1, 2019,new Date()),
                        new VideoOrder("1", "linux ", 2, 2019,new Date()),
                        new VideoOrder("1", "db ", 3, 2019,new Date()),
                        new VideoOrder("1", "java ", 1, 2019,new Date()),
                        new VideoOrder("1", "linux ", 2, 2019,new Date()),
                        new VideoOrder("1", "db ", 3, 2019,new Date())
                )
        );

        //过滤  <T, O>  T是参数类型 O是返回的参数类型
        SingleOutputStreamOperator<VideoOrder> filterDs = ds.flatMap(new FlatMapFunction<VideoOrder, VideoOrder>() {
            @Override
            public void flatMap(VideoOrder value, Collector<VideoOrder> out) throws Exception {
                if (Objects.nonNull(value) && NumberUtils.INTEGER_ONE.equals(value.getMoney())) {
                    //去掉money不是1的数据
                    out.collect(value);
                }
            }
        });
        //看  keySelector 中的泛型为<IN, KEY>  IN 输入值类型 和  分组KEY的类型
        KeyedStream<VideoOrder, String> keyByDs = filterDs.keyBy(new KeySelector<VideoOrder, String>() {
            @Override
            public String getKey(VideoOrder value) throws Exception {
                return  value.getTitle();
            }
        });
        //求和的字段属性是  money
        SingleOutputStreamOperator<VideoOrder> sumDs = keyByDs.sum("money");
        sumDs.print();
        env.execute("test");
    }
}
